// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
//
// Author: wsfuyibing <682805@qq.com>
// Date: 2024-12-05

package supporter

import (
	"gitee.com/gomq/sr/endpoint"
	"gitee.com/gomq/sr/errs"
	"gitee.com/gomq/sr/scheduler"
	"gitee.com/gomq/sr/servlet"
	"gitee.com/gomq/sr/transform"
	"sync"
)

var (
	once      = new(sync.Once)
	singleton Supporter
)

type (
	// Supporter is an interface used combine message queue middleware with
	// consume dispatcher.
	Supporter interface {
		AddScheduler(builder scheduler.Builder, protocols ...endpoint.Protocol) (err error)
		AddServlet(builder servlet.Builder, plugins ...servlet.Plugin) (err error)
		AddTransform(builder transform.Builder, transitions ...transform.Transition) (err error)
		DelScheduler(protocol endpoint.Protocol)
		DelServlet(plugin servlet.Plugin)
		DelTransform(transition transform.Transition)
		HasScheduler(protocol endpoint.Protocol) (yes bool)
		HasServlet(plugin servlet.Plugin) (yes bool)
		HasTransform(transition transform.Transition) (yes bool)
		ListSchedulers() map[endpoint.Protocol]scheduler.Builder
		ListServlets() map[servlet.Plugin]servlet.Builder
		ListTransforms() map[transform.Transition]transform.Builder
		NewScheduler(protocol endpoint.Protocol) (scheduler scheduler.Scheduler, err error)
		NewServlet(plugin servlet.Plugin) (servlet servlet.Servlet, err error)
		NewTransform(transition transform.Transition) (transform transform.Transform, err error)
	}

	supporter struct {
		schedulers  map[endpoint.Protocol]scheduler.Builder
		servlets    map[servlet.Plugin]servlet.Builder
		transitions map[transform.Transition]transform.Builder
		mu          *sync.RWMutex
	}
)

// Singleton
// return the singleton instance of supporter.
func Singleton() Supporter { return singleton }

func (o *supporter) AddScheduler(builder scheduler.Builder, protocols ...endpoint.Protocol) (err error) {
	o.mu.Lock()
	defer o.mu.Unlock()

	for _, protocol := range protocols {
		if protocol.Undefined() {
			continue
		}
		if _, ok := o.schedulers[protocol]; ok {
			err = errs.ErrSchedulerHasExist
			break
		}
		o.schedulers[protocol] = builder
	}
	return
}

func (o *supporter) AddServlet(builder servlet.Builder, plugins ...servlet.Plugin) (err error) {
	o.mu.Lock()
	defer o.mu.Unlock()

	for _, plugin := range plugins {
		if plugin.Undefined() {
			continue
		}
		if _, ok := o.servlets[plugin]; ok {
			err = errs.ErrServletHasExist
			break
		}
		o.servlets[plugin] = builder
	}
	return
}

func (o *supporter) AddTransform(builder transform.Builder, transitions ...transform.Transition) (err error) {
	o.mu.Lock()
	defer o.mu.Unlock()

	for _, transition := range transitions {
		if transition.Undefined() {
			continue
		}
		if _, ok := o.transitions[transition]; ok {
			err = errs.ErrTransformHasExist
			break
		}
		o.transitions[transition] = builder
	}
	return
}

func (o *supporter) DelScheduler(protocol endpoint.Protocol) {
	o.mu.Lock()
	delete(o.schedulers, protocol)
	o.mu.Unlock()
}

func (o *supporter) DelServlet(plugin servlet.Plugin) {
	o.mu.Lock()
	delete(o.servlets, plugin)
	o.mu.Unlock()
}

func (o *supporter) DelTransform(transition transform.Transition) {
	o.mu.Lock()
	delete(o.transitions, transition)
	o.mu.Unlock()
}

func (o *supporter) HasScheduler(protocol endpoint.Protocol) (yes bool) {
	o.mu.RLock()
	defer o.mu.RUnlock()
	_, yes = o.schedulers[protocol]
	return
}

func (o *supporter) HasServlet(plugin servlet.Plugin) (yes bool) {
	o.mu.RLock()
	defer o.mu.RUnlock()
	_, yes = o.servlets[plugin]
	return
}

func (o *supporter) HasTransform(transition transform.Transition) (yes bool) {
	o.mu.RLock()
	defer o.mu.RUnlock()
	_, yes = o.transitions[transition]
	return
}

func (o *supporter) ListSchedulers() map[endpoint.Protocol]scheduler.Builder {
	o.mu.RLock()
	defer o.mu.RUnlock()
	return o.schedulers
}

func (o *supporter) ListServlets() map[servlet.Plugin]servlet.Builder {
	o.mu.RLock()
	defer o.mu.RUnlock()
	return o.servlets
}

func (o *supporter) ListTransforms() map[transform.Transition]transform.Builder {
	o.mu.RLock()
	defer o.mu.RUnlock()
	return o.transitions
}

func (o *supporter) NewScheduler(protocol endpoint.Protocol) (scheduler scheduler.Scheduler, err error) {
	o.mu.RLock()
	defer o.mu.RUnlock()

	if builder, ok := o.schedulers[protocol]; ok {
		if scheduler = builder(); scheduler == nil {
			err = errs.ErrSchedulerNotBuilt
		}
		return
	}

	err = errs.ErrSchedulerNotExist
	return
}

func (o *supporter) NewServlet(plugin servlet.Plugin) (servlet servlet.Servlet, err error) {
	o.mu.RLock()
	defer o.mu.RUnlock()

	if builder, ok := o.servlets[plugin]; ok {
		if servlet = builder(); servlet == nil {
			err = errs.ErrServletNotBuilt
		}
		return
	}

	err = errs.ErrServletNotExist
	return
}

func (o *supporter) NewTransform(transition transform.Transition) (transform transform.Transform, err error) {
	o.mu.RLock()
	defer o.mu.RUnlock()

	if builder, ok := o.transitions[transition]; ok {
		if transform = builder(); transform == nil {
			err = errs.ErrTransformNotBuilt
		}
		return
	}

	err = errs.ErrTransformNotExist
	return
}

func (o *supporter) init() *supporter {
	o.mu = new(sync.RWMutex)
	o.schedulers = make(map[endpoint.Protocol]scheduler.Builder)
	o.servlets = make(map[servlet.Plugin]servlet.Builder)
	o.transitions = make(map[transform.Transition]transform.Builder)
	return o
}

func init() {
	once.Do(func() {
		singleton = (&supporter{}).init()
	})
}
